package com.jahe.learn.nio.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *  多线程Reactor反应器的实践案例
 * @author: jahe.lai
 * @create: 2019-12-24 14:29
 */
public class MultiThreadEchoServerReactor {

    ServerSocketChannel serverSocketChannel;
    AtomicInteger next = new AtomicInteger(0);
    // 选择器集合，引入多个选择器
    Selector[] selectors = new Selector[2];
    // 引入多个子反应器
    SubReactor[] subReactors;

    MultiThreadEchoServerReactor() throws IOException {
        selectors[0] = Selector.open();
        selectors[1] = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 8899);
        serverSocketChannel.socket().bind(address);
        // 非阻塞
        serverSocketChannel.configureBlocking(false);
        // 第一个选择器，负责监听新连接事件
        SelectionKey skey = serverSocketChannel.register(selectors[0], SelectionKey.OP_ACCEPT);
        // skey绑定 新连接到来时的 hander处理器对象
        skey.attach(new AcceptHandler());
        // 第一个子反应器，负责一个选择器
        SubReactor subReactor0 = new SubReactor(selectors[0]);
        // 第二个子反应器，负责另一个选择器
        SubReactor subReactor1 = new SubReactor(selectors[1]);
        subReactors = new SubReactor[] { subReactor0, subReactor1 };
    }

    public void startService() {
        // 一个子反应器， 对应一个线程
        new Thread(subReactors[0]).start();
        new Thread(subReactors[1]).start();
    }
    public static void main(String[] args) throws IOException {
        MultiThreadEchoServerReactor serverReactor = new MultiThreadEchoServerReactor();
        serverReactor.startService();
    }


    // 连接处理器
    class AcceptHandler implements Runnable {

        @Override
        public void run() {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                if (null != socketChannel) {
                    // 连接成功后的处理逻辑
                    new MultiThreadEchoHandler(socketChannel, selectors[next.get()]);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            if (next.incrementAndGet() == selectors.length) {
                next.set(0);
            }
        }
    }
}


// 子反应器
class SubReactor implements Runnable {

    // 每个线程负责一个子反应器的查询 和 选择
    final  Selector selector;
    public SubReactor(Selector selector) {
        this.selector = selector;
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                selector.select();
                Set<SelectionKey> keySet = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = keySet.iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    dispatch(key);
                }
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    void dispatch(SelectionKey key) {
        Runnable hander = (Runnable)key.attachment();
        // 调用之前绑定到 SelectionKey 上的handler处理器对象
        if (null != hander) {
            hander.run();
        }
    }
}

class MultiThreadEchoHandler implements Runnable {

    final SocketChannel socketChannel;
    final SelectionKey skey;
    final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    static final int RECIEVING = 0, SENDING = 1;
    int state = RECIEVING;

    // 引入线程池
    static ExecutorService pool = Executors.newFixedThreadPool(4);

    MultiThreadEchoHandler(SocketChannel channel, Selector selector) throws IOException {
        this.socketChannel = channel;
        this.socketChannel.configureBlocking(false);
        // 取得选择键， 后面设置感兴趣的事件类型
        this.skey = this.socketChannel.register(selector, 0);
        // 将本对象作为 SelectionKey 的附件，方便事件分发。
        this.skey.attach(this);
        // 注册 READ 事件
        this.skey.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }

    @Override
    public void run() {
        // 当前线程执行任务
//        new AsyncTask().run();
        // 在线程池执行任务。当前线程可直接返回
        pool.execute(new AsyncTask());
    }

    public synchronized void asyncRun() {
        try {
            if (state == SENDING) {
                // 写入通道
                socketChannel.write(byteBuffer);
                byteBuffer.clear();
                // 写完之后，注册为 读取事件
                skey.interestOps(SelectionKey.OP_READ);
                // 进入接收的状态
                state = RECIEVING;
            } else if (state == RECIEVING){
                int lenth = 0;
                // 从通道读取
                while ((lenth = socketChannel.read(byteBuffer)) > 0) {
                    System.out.println(new String(byteBuffer.array(), 0, lenth));
                }
                // 读取完成后。bytebuffer 切换到 写模式
                byteBuffer.flip();
                // 切换到 监听 写出 事件
                skey.interestOps(SelectionKey.OP_WRITE);
                // 读取完成后，进入发送状态
                state = SENDING;
            }
            // 处理结束 了，这里不能关闭 SelectionKey, 因为监听了多个事件，重复使用。
            // skey.cancel();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 异步任务的内部类
    class AsyncTask implements Runnable {
        @Override
        public void run() {
            MultiThreadEchoHandler.this.asyncRun();
        }
    }
}
